package com.mlog.njyjy.ndop.share.kafka.oauth.handler;

import com.mlog.njyjy.ndop.share.kafka.oauth.OauthRequest;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
import org.apache.kafka.common.utils.Time;

import javax.security.auth.callback.Callback;
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.auth.login.AppConfigurationEntry;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * <p>
 * .
 * </p>
 *
 * @author <a href="mailto:yingxiufeng@mlogcn.com">alex</a>
 * @version v1.0, 2020/7/17
 */
public abstract class OauthAuthenticateBaseCallbackHandler implements AuthenticateCallbackHandler {

    protected Map<String, String> moduleOptions = null;

    protected boolean configured = false;

    protected OauthRequest oauthRequest = OauthRequest.create();

    /**
     * Configures this callback handler for the specified SASL mechanism.
     *
     * @param configs           Key-value pairs containing the parsed configuration options of
     *                          the client or broker. Note that these are the Kafka configuration options
     *                          and not the JAAS configuration options. JAAS config options may be obtained
     *                          from `jaasConfigEntries` for callbacks which obtain some configs from the
     *                          JAAS configuration. For configs that may be specified as both Kafka config
     *                          as well as JAAS config (e.g. sasl.kerberos.service.name), the configuration
     *                          is treated as invalid if conflicting values are provided.
     * @param saslMechanism     Negotiated SASL mechanism. For clients, this is the SASL
     *                          mechanism configured for the client. For brokers, this is the mechanism
     *                          negotiated with the client and is one of the mechanisms enabled on the broker.
     * @param jaasConfigEntries JAAS configuration entries from the JAAS login context.
     *                          This list contains a single entry for clients and may contain more than
     *                          one entry for brokers if multiple mechanisms are enabled on a listener using
     *                          static JAAS configuration where there is no mapping between mechanisms and
     *                          login module entries. In this case, callback handlers can use the login module in
     *                          `jaasConfigEntries` to identify the entry corresponding to `saslMechanism`.
     *                          Alternatively, dynamic JAAS configuration option
     *                          {@link SaslConfigs#SASL_JAAS_CONFIG} may be
     *                          configured on brokers with listener and mechanism prefix, in which case
     *                          only the configuration entry corresponding to `saslMechanism` will be provided
     */
    @Override
    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
        if (!OAuthBearerLoginModule.OAUTHBEARER_MECHANISM.equals(saslMechanism)) {
            throw new IllegalArgumentException(String.format("Unexpected SASL mechanism: %s", saslMechanism));
        }
        if (Objects.requireNonNull(jaasConfigEntries).size() != 1 || jaasConfigEntries.get(0) == null) {
            throw new IllegalArgumentException(
                    String.format("Must supply exactly 1 non-null JAAS mechanism configuration (size was %d)",
                            jaasConfigEntries.size()));
        }
        this.moduleOptions = Collections.unmodifiableMap((Map<String, String>) jaasConfigEntries.get(0).getOptions());
        configured = true;
    }

    /**
     * Closes this instance.
     */
    @Override
    public void close() {

    }

    /**
     * <p> Retrieve or display the information requested in the
     * provided Callbacks.
     *
     * <p> The {@code handle} method implementation checks the
     * instance(s) of the {@code Callback} object(s) passed in
     * to retrieve or display the requested information.
     * The following example is provided to help demonstrate what an
     * {@code handle} method implementation might look like.
     * This example code is for guidance only.  Many details,
     * including proper error handling, are left out for simplicity.
     *
     * <pre>{@code
     * public void handle(Callback[] callbacks)
     * throws IOException, UnsupportedCallbackException {
     *
     *   for (int i = 0; i < callbacks.length; i++) {
     *      if (callbacks[i] instanceof TextOutputCallback) {
     *
     *          // display the message according to the specified type
     *          TextOutputCallback toc = (TextOutputCallback)callbacks[i];
     *          switch (toc.getMessageType()) {
     *          case TextOutputCallback.INFORMATION:
     *              System.out.println(toc.getMessage());
     *              break;
     *          case TextOutputCallback.ERROR:
     *              System.out.println("ERROR: " + toc.getMessage());
     *              break;
     *          case TextOutputCallback.WARNING:
     *              System.out.println("WARNING: " + toc.getMessage());
     *              break;
     *          default:
     *              throw new IOException("Unsupported message type: " +
     *                                  toc.getMessageType());
     *          }
     *
     *      } else if (callbacks[i] instanceof NameCallback) {
     *
     *          // prompt the user for a username
     *          NameCallback nc = (NameCallback)callbacks[i];
     *
     *          // ignore the provided defaultName
     *          System.err.print(nc.getPrompt());
     *          System.err.flush();
     *          nc.setName((new BufferedReader
     *                  (new InputStreamReader(System.in))).readLine());
     *
     *      } else if (callbacks[i] instanceof PasswordCallback) {
     *
     *          // prompt the user for sensitive information
     *          PasswordCallback pc = (PasswordCallback)callbacks[i];
     *          System.err.print(pc.getPrompt());
     *          System.err.flush();
     *          pc.setPassword(readPassword(System.in));
     *
     *      } else {
     *          throw new UnsupportedCallbackException
     *                  (callbacks[i], "Unrecognized Callback");
     *      }
     *   }
     * }
     *
     * // Reads user password from given input stream.
     * private char[] readPassword(InputStream in) throws IOException {
     *    // insert code to read a user password from the input stream
     * }
     * }</pre>
     *
     * @param callbacks an array of {@code Callback} objects provided
     *                  by an underlying security service which contains
     *                  the information requested to be retrieved or displayed.
     * @throws IOException                  if an input or output error occurs. <p>
     * @throws UnsupportedCallbackException if the implementation of this
     *                                      method does not support one or more of the Callbacks
     *                                      specified in the {@code callbacks} parameter.
     */
    @Override
    public abstract void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException;
}
